package com.atguigu.sparkstreaming.demos

import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.kafka010._
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * Created by Smexy on 2022/5/23
 *
Caused by: java.io.NotSerializableException:
    Object of org.apache.spark.streaming.kafka010.DirectKafkaInputDStream
      is being serialized  possibly
      as a part of closure of an RDD operation

    DirectKafkaInputDStream 无法序列化，无法闭包。


 *
 *
 */
object CommitOffsetsDemo1 {

  def main(args: Array[String]): Unit = {

    val streamingContext = new StreamingContext("local[*]", "SparkStreamingKafkaDemo", Seconds(5))


    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "hadoop102:9092,hadoop103:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "atguigu1227",
      "auto.offset.reset" -> "latest",
      // ①取消offsets的自动提交
      "enable.auto.commit" -> "false"
    )


    val topics1 = Array("topicA")


    // DirectKafkaInputDStream 里面封装的才是KafkaRDD
    val stream = KafkaUtils.createDirectStream[String, String](
      streamingContext,
      LocationStrategies.PreferConsistent,
      ConsumerStrategies.Subscribe[String, String](topics1, kafkaParams)
    )

    stream.foreachRDD(rdd => {

      //获取偏移量
      val offsetRanges: Array[OffsetRange] = rdd.asInstanceOf[HasOffsetRanges].offsetRanges

      //输出方法
      rdd.foreach(record => {
        println(record+":"+Thread.currentThread().getName)

        //1.不能提交  ①无法实现  ②实现逻辑上，当前Task不能提交全部分区偏移量
        //stream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)
      })

      //2提交
      stream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)

    })

    // 启动app
    streamingContext.start()

    // 阻塞当前线程，让程序24h不停运行
    streamingContext.awaitTermination()


  }

}
